package com.shujia.tour

import org.apache.spark.internal.Logging
import org.apache.spark.sql._
object ScenicTouristApp extends Logging {

  def main(args: Array[String]): Unit = {


    if (args.length == 0) {
      log.error("输出参数为空")
      return
    }

    val day_id = args(0)

    log.info(s"当前天时间分区为：$day_id")

    val month_id: String = day_id.substring(0, 6)

    log.info(s"当前月时间分区为：$month_id")

    val spark: SparkSession = SparkSession.builder()
      .appName("CityTouristWideApp")
      .config("spark.sql.shuffle.partitions", "20")
      .enableHiveSupport()
      .getOrCreate()



    /**
      *
      * 景区游客过滤
      * 1、目的地在景区内
      * 2、常住地不在景区内
      *
      *
      */


   val resulyDF: DataFrame =  spark.sql(
      s"""
        |SELECT
        | f.mdn,
        | j.city_id AS source_city_id,
        | f.scenic_id AS d_scenic_id,
        | f.scenic_name AS d_scenic_name,
        | min(f.grid_first_time) as d_arrive_time,
        | sum(f.duration) as d_stay_time
        |FROM
        | (
        |  SELECT
        |  /*+ BROADCAST(a) */
        |   a.scenic_id,
        |   a.scenic_name,
        |   b.mdn,
        |   b.county_id,
        |   b.duration,
        |   b.grid_first_time
        |  FROM
        |   dim.dim_scenic_grid AS a
        |  INNER JOIN (
        |   SELECT
        |    *
        |   FROM
        |    dwi.dwi_staypoint_msk_d
        |   WHERE
        |    day_id = $day_id
        |  ) AS b ON a.grid = b.grid_id
        | ) AS f
        |INNER JOIN (
        | SELECT
        | /*+ BROADCAST(e) */
        |  d.mdn,
        |  e.city_id
        | FROM
        |  (
        |   SELECT
        |   /*+ BROADCAST(c) */
        |    mdn,
        |    resi_county_id
        |   FROM
        |    (
        |     SELECT
        |      mdn,
        |      resi_grid_id,
        |      resi_county_id
        |     FROM
        |      dim.dim_usertag_msk_m
        |     WHERE
        |      month_id = $month_id
        |    ) AS d
        |   LEFT JOIN dim.dim_scenic_grid AS c ON c.grid = d.resi_grid_id
        |   WHERE
        |    c.grid IS NULL
        |  ) AS d
        | JOIN dim.dim_admincode AS e ON d.resi_county_id = e.county_id
        |) AS j ON f.mdn = j.mdn
        |GROUP BY
        | f.mdn,
        | j.city_id,
        | f.scenic_id,
        | f.scenic_name
        |
        |
      """.stripMargin)


    resulyDF
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dal_tour/dal_tour_scenic_tourist_msk_d/day_id=$day_id")

    spark.sql(s"alter table  dal_tour.dal_tour_scenic_tourist_msk_d add if not exists partition(day_id='$day_id')")


  }

}
